package com.gzsxy.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author xiaolong
 * @version 1.0
 * @description: 生产者异步回调方式发送消息
 * @date 2022/7/5 21:00
 */
public class MySimpleProducer1 {

    private final static String TOPIC_NAME = "test";

    public static void main(String[] args) throws ExecutionException,InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.113.229.173:9092");


        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //发消息的客户端
        Producer<String, String> producer = new KafkaProducer<String, String>(props);


        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME,
                "mykey", "hello");


        //异步回调方式发送消息
        producer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("发送消息失败：" +
                            exception.getStackTrace());
                }
                if (metadata != null) {
                    System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
                }
            }
        });

        Thread.sleep(1000000L);
    }


}
